package top.ruanlitao.shortlink.project.mq.redis.consumer;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import top.ruanlitao.shortlink.project.common.constant.RedisKeyConstant;
import top.ruanlitao.shortlink.project.common.convention.exception.ServiceException;
import top.ruanlitao.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import top.ruanlitao.shortlink.project.mq.idempotent.MessageQueueIdempotentHandler;
import top.ruanlitao.shortlink.project.service.ShortLinkService;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 延迟消费短链接统计数据 消费者
 */
@ConditionalOnProperty(prefix = "short-link.message-queue", name = "type", havingValue = "redis")
@Component
@RequiredArgsConstructor
@Slf4j
@Deprecated
public class DelayShortLinkStatsConsumer implements InitializingBean {

    private final RedissonClient redissonClient;
    private final ShortLinkService shortLinkService;
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;

    public void onMessage() {
        // 创建一个单线程的线程池执行器
        ExecutorService executorService = Executors.newSingleThreadExecutor(runnable -> {
            // 通过自定义线程工厂创建线程
            Thread thread = new Thread(runnable);
            thread.setName("delay_short-link_stats_consumer");
            // 将线程设置为守护线程，意味着当主程序退出时，这个线程也会自动退出。
            thread.setDaemon(Boolean.TRUE);
            return thread;
        });
        // 执行任务
        executorService.execute(() -> {
            // 获取 Redis 延迟队列
            RBlockingDeque<ShortLinkStatsRecordDTO> blockingDeque = redissonClient.getBlockingDeque(RedisKeyConstant.DELAY_QUEUE_STATS_KEY);
            RDelayedQueue<ShortLinkStatsRecordDTO> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
            // 无限循环消费队列中的数据
            for(; ; ) {
                try {
                    // 获取消息记录
                    ShortLinkStatsRecordDTO statsRecord =  blockingDeque.take();
                    // 幂等性判断
                    if (messageQueueIdempotentHandler.isMessageBeingConsumed(statsRecord.getKeys())) {
                        if (messageQueueIdempotentHandler.isAccomplish(statsRecord.getKeys())) {
                            return;
                        }
                        throw new ServiceException("消息未完成流程，需要消息队列重试");
                    }
                    try {
                        log.info("短链接 {} 监控数据尝试加入数据库了", statsRecord.getFullShortUrl());
                        shortLinkService.shortLinkStats(statsRecord);
                    } catch (Throwable ex) {
                        messageQueueIdempotentHandler.delMessageProcessed(statsRecord.getKeys());
                        log.error("延迟记录短链接监控消费异常", ex);
                    }
                    messageQueueIdempotentHandler.setAccomplish(statsRecord.getKeys());
                } catch (Throwable ignored) {}
            }
        });
    }

    @Override
    public void afterPropertiesSet() {
        onMessage();
    }
}
